#include "StdAfx.h"
#include "DistributeDataTask.h"
#include "..\VideoAsynchEncoded\synchronization\CriticalSection.h"
#include "..\Srv_Iocp_lib\Srv_Service_Handler.h"

CDistributeDataTask::CDistributeDataTask(void)
{
	m_pSocketCriticalSection   = new CCriticalSection;
}

CDistributeDataTask::~CDistributeDataTask(void)
{
	close();
	delete m_pSocketCriticalSection;
}

int CDistributeDataTask::svc()
{
 	std::string str ="";
 	while(getq(str))
 	{
 		if (str=="")
 			return true;
 
 		
 		{
 			CCriticalAutoLock loAutoLock(*m_pSocketCriticalSection);
 
 			std::map<int,Srv_Service_Handler*> ::iterator  it = map_All.begin();
 			while(it != map_All.end())
 			{
 				it->second->Stream_write(str.data(),str.length());
 				it ++;
 			}
 		}
 	}
 	return 1;
}

void CDistributeDataTask::add( Srv_Service_Handler* p_socket_handle )
{
	if (p_socket_handle)
	{
		CCriticalAutoLock loAutoLock(*m_pSocketCriticalSection);
		map_All[p_socket_handle->socket()] = p_socket_handle;
	}
	CDistributeDataTask::Interface().activate();
}

void CDistributeDataTask::del( Srv_Service_Handler* p_socket_handle)
{
	if (p_socket_handle)
	{
		CCriticalAutoLock loAutoLock(*m_pSocketCriticalSection);
		std::map<int,Srv_Service_Handler*> ::iterator  it =map_All.find(p_socket_handle->socket());
		if (it != map_All.end())
		{
			map_All.erase(it);
		}
	}
}

int CDistributeDataTask::close()
{
	std::string str("");
	int nthread  =thr_count();

	for (int i =0 ;i<nthread ;i++)
		putq(str);

	wait();
	return 0;
}

bool CDistributeDataTask::receive_handler( char *data,int size )
{
	CCriticalAutoLock loAutoLock(*m_pSocketCriticalSection);
	//std::string str_buf;

	//str_buf.append(data,size);				

	putq(std::string(data,size));		
	return true;
}
CDistributeDataTask DistributeTask;

CDistributeDataTask & CDistributeDataTask::Interface()
{
	return  DistributeTask;
}

